package leafnode

//wrap all the replication operations
import (
	"errors"
	"net"
	"strconv"
	"sync/atomic"
)

func (s *Server) initFollowerChan(size int) {
	s.followerChan = make(chan *FollowerCohort, size)
	fc := NewFollowerCohort()

	//enqueue only one
	s.followerChan <- fc
	atomic.AddInt32(&s.followerConnections, 1)
}

func (s *Server) totalFollowerConnections() int32 {
	return atomic.LoadInt32(&s.followerConnections)
}

func (s *Server) shutFollowerConn(fc *FollowerCohort) {
	for _, f := range fc.followers {
		f.Shutdown()
	}

	atomic.AddInt32(&s.followerConnections, -1)
}

func (s *Server) createFollowerConn() (fc *FollowerCohort, e error) {
	fc = NewFollowerCohort()
	for j, f := range fc.followers {
		//check connection
		e = f.Setup(s.followerAddr[j])
		if e != nil {
			s.errLog.Println("Error: failed to set up follower ", f.addr, e)
			break
		}
	}

	if e == nil {
		atomic.AddInt32(&s.followerConnections, 1)
		return
	}

	for _, f := range fc.followers {
		f.Shutdown()
	}
	return nil, e
}

func (s *Server) getFollowerConn() (fc *FollowerCohort, e error) {
	select {
	case fc = <-s.followerChan:
		return
	default:
		fc, e = s.createFollowerConn()
	}

	return
}

func (s *Server) putFollowerConn(fc *FollowerCohort) {
	select {
	case s.followerChan <- fc:
	default:
		s.shutFollowerConn(fc)
	}
}

type FollowerCohort struct {
	followers [FullReplFactor]*Follower
}

func NewFollowerCohort() *FollowerCohort {
	fc := new(FollowerCohort)
	for i := range fc.followers {
		fc.followers[i] = NewFollower()
	}
	return fc
}

//the follower abstraction
type Follower struct {
	addr string
	conn *net.TCPConn
}

func NewFollower() *Follower {
	return new(Follower)
}

func (f *Follower) Disconnected() bool {
	return f.conn == nil
}

func (f *Follower) Reconnect() error {
	//TODO - better re-connect mechanism
	//<-time.After(ReconnectDelay * time.Millisecond)
	return f.Setup(f.addr)
}

func (f *Follower) Setup(addr string) error {
	//firstly close it
	f.Shutdown()

	f.addr = addr
	rc, e := net.Dial("tcp", f.addr)
	if e == nil {
		var ok bool
		f.conn, ok = rc.(*net.TCPConn)

		if !ok {
			rc.Close()
			return errors.New("tcp conn convert failed")
		}

		//set it nodelay & keepalive
		f.conn.SetNoDelay(true)
		f.conn.SetKeepAlive(true)
	}

	return e
}

func (f *Follower) Shutdown() {
	if f.conn != nil {
		f.conn.Close()
		f.conn = nil
	}
}

func (f *Follower) RelayWriteCmd(data []byte, chunk int, initialOffset, totalSize, offset, length int64) error {
	if f.conn == nil {
		return errors.New("DisconnectedFollower")
	}
	p := NewPacket()
	p.PackReplWriteRequest(chunk, initialOffset, totalSize, offset, length, data)
	err := p.WriteToConn(f.conn)
	if err != nil {
		f.Shutdown()
	}
	return err
}

func (f *Follower) FetchWriteReply() error {
	if f.conn == nil {
		return errors.New("DisconnectedFollower")
	}
	p := NewPacket()
	err := p.ReadFromConn(f.conn)
	if err != nil {
		f.Shutdown()
	}

	if p.Opcode != OkReplWrite {
		err = errors.New(string(p.Args[0]))
	}
	return err
}

func (f *Follower) GetWatermark(chunkId int) (offset int64, e error) {
	if f.conn == nil {
		e = errors.New("DisconnectedFollower")
		return
	}
	req := NewPacket()
	req.Opcode = Watermark
	req.Argnum = 1
	req.Args[0] = []byte(strconv.Itoa(chunkId))
	req.Arglength[0] = uint32(len(req.Args[0]))

	e = req.WriteToConn(f.conn)
	if e != nil {
		f.Shutdown()
		return
	}

	res := NewPacket()
	e = res.ReadFromConn(f.conn)
	if e != nil {
		f.Shutdown()
		return
	}

	if res.Opcode == OkWatermark {
		offset, e = strconv.ParseInt(string(res.Args[0]), 10, 64)
		return
	}

	e = errors.New(string(res.Args[0]))
	return
}

//fetch data from the follower, length <= 4MB
func (f *Follower) SyncFrom(chunkId int, offset, length int64) (data []byte, e error) {
	if f.conn == nil {
		e = errors.New("DisconnectedFollower")
		return
	}

	//pack the request
	req := NewPacket()
	req.Argnum = 1
	req.Opcode = SyncFrom
	inkey := strconv.Itoa(chunkId) + SplitString +
		strconv.FormatInt(offset, 10) + SplitString + strconv.FormatInt(length, 10)
	req.Args[0] = []byte(inkey)
	req.Arglength[0] = uint32(len(req.Args[0]))

	//send it
	e = req.WriteToConn(f.conn)
	if e != nil {
		f.Shutdown()
		return
	}

	//fetch response
	res := NewPacket()
	e = res.ReadFromConn(f.conn)
	if e != nil {
		f.Shutdown()
		return
	}

	if res.Opcode == OkSyncFrom {
		data = res.Args[0]
	} else {
		e = errors.New(string(res.Args[0]))
	}
	return
}

//send data to the follower, length <= 4MB
func (f *Follower) SyncTo(chunkId int, offset, length int64, data []byte) error {
	if f.conn == nil {
		return errors.New("DisconnectedFollower")
	}
	req := NewPacket()
	req.Argnum = 2
	req.Opcode = SyncTo
	inkey := strconv.Itoa(chunkId) + SplitString +
		strconv.FormatInt(offset, 10) + SplitString + strconv.FormatInt(length, 10)
	req.Args[0] = []byte(inkey)
	req.Arglength[0] = uint32(len(req.Args[0]))
	req.Args[1] = data
	req.Arglength[1] = uint32(len(data))
	e := req.WriteToConn(f.conn)
	if e != nil {
		f.Shutdown()
		return e
	}
	res := NewPacket()
	e = res.ReadFromConn(f.conn)
	if e != nil {
		f.Shutdown()
		return e
	}
	if res.Opcode != OkSyncTo {
		e = errors.New(string(res.Args[0]))
	}
	return e
}
